package com.example.demo;

import com.alibaba.fastjson.JSONObject;
import com.example.demo.modules.entity.AlarmHistoryLog;
import com.example.demo.modules.entity.DeviceInfo;
import com.example.demo.modules.entity.DevicePropHistoryLog;
import com.example.demo.modules.service.IAlarmHistoryLogService;
import com.example.demo.modules.service.IDeviceInfoService;
import com.example.demo.modules.service.IDevicePropHistoryLogService;
import org.apache.commons.codec.binary.Base64;
import org.apache.qpid.jms.JmsConnection;
import org.apache.qpid.jms.JmsConnectionListener;
import org.apache.qpid.jms.message.JmsInboundMessageDispatch;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;

import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import java.net.URI;
import java.time.LocalDateTime;
import java.util.Hashtable;

@Configuration
public class DeviceMessage implements InitializingBean {

    @Value("${user.accessKeyId}")
    private String accessKeyId;

    @Value("${user.accessKeySecret}")
    private String accessKeySecret;

    @Value("${user.uid}")
    private String uid;

    @Value("${iot.regionId}")
    private String regionId;

    @Value("${iot.instance}")
    private String instance;

    @Autowired
    IDeviceInfoService deviceInfoService;

    @Autowired
    IDevicePropHistoryLogService devicePropHistoryLogService;

    @Autowired
    IAlarmHistoryLogService alarmHistoryLogService;

    @Override
    public void afterPropertiesSet() throws Exception {

        long timeStamp = System.currentTimeMillis();
        //签名方法：支持hmacmd5、hmacsha1和hmacsha256。
        String signMethod = "hmacsha1";

        //userName组装方法，请参见AMQP客户端接入说明文档。
        String userName = "ac:de:48:00:11:22" + "-" + 0 + "|authMode=aksign"
                + ",signMethod=" + signMethod
                + ",timestamp=" + timeStamp
                + ",authId=" + accessKeyId
                + ",iotInstanceId=" + instance
                + ",consumerGroupId=" + "DEFAULT_GROUP"
                + "|";
        //计算签名，password组装方法，请参见AMQP客户端接入说明文档。
        String signContent = "authId=" + accessKeyId + "&timestamp=" + timeStamp;
        String password = doSign(signContent, accessKeySecret, signMethod);
        String host = uid + ".iot-amqp." + regionId + ".aliyuncs.com";
        String connectionUrl = "failover:(amqps://" + host + ":5671?amqp.idleTimeout=80000)"
                + "?failover.reconnectDelay=30";

        Hashtable<String, String> hashtable = new Hashtable<>();
        hashtable.put("connectionfactory.SBCF", connectionUrl);
        hashtable.put("queue.QUEUE", "default");
        hashtable.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.qpid.jms.jndi.JmsInitialContextFactory");
        Context context = new InitialContext(hashtable);
        ConnectionFactory cf = (ConnectionFactory)context.lookup("SBCF");
        Destination queue = (Destination)context.lookup("QUEUE");
        // 创建连接。
        Connection connection = cf.createConnection(userName, password);
//        connections.add(connection);

        ((JmsConnection)connection).addConnectionListener(myJmsConnectionListener);
        // 创建会话。
        // Session.CLIENT_ACKNOWLEDGE: 收到消息后，需要手动调用message.acknowledge()。
        // Session.AUTO_ACKNOWLEDGE: SDK自动ACK（推荐）。
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        connection.start();
        MessageListener messageListener = new MessageListener() {
            @Override
            public void onMessage(final Message message) {
                try {
                    //1.收到消息之后一定要ACK。
                    // 推荐做法：创建Session选择Session.AUTO_ACKNOWLEDGE，这里会自动ACK。
                    // 其他做法：创建Session选择Session.CLIENT_ACKNOWLEDGE，这里一定要调message.acknowledge()来ACK。
                    // message.acknowledge();
                    //2.建议异步处理收到的消息，确保onMessage函数里没有耗时逻辑。
                    // 如果业务处理耗时过程过长阻塞住线程，可能会影响SDK收到消息后的正常回调。
                    processMessage(message);
                } catch (Exception e) {
                    System.out.println("submit task occurs exception " + e);
                }
            }
        };

        // 创建Receiver连接。
        MessageConsumer consumer = session.createConsumer(queue);
        consumer.setMessageListener(messageListener);

    }

    /**
     * 在这里处理您收到消息后的具体业务逻辑。
     */
    private void processMessage(Message message) {
        try {
            byte[] body = message.getBody(byte[].class);
            String content = new String(body);
            String topic = message.getStringProperty("topic");
            String messageId = message.getStringProperty("messageId");
            System.out.println("receive message"
                    + ",\n topic = " + topic
                    + ",\n messageId = " + messageId
                    + ",\n content = " + content);

            // 接收到设备上下线消息以后的处理
            if (topic.contains("/as/mqtt/status/")) {
                JSONObject jsonObject = JSONObject.parseObject(content); // 字符串转json对象
                DeviceInfo deviceInfo = new DeviceInfo();
                deviceInfo.setDeviceName(jsonObject.getString("deviceName"));
                deviceInfo.setConnectStatus(jsonObject.getString("status").equals("online"));
                if (jsonObject.getString("status").equals("online")) {
                    deviceInfo.setLastOnlineDate(LocalDateTime.now());
                }
                deviceInfoService.saveDeviceInfo(deviceInfo);
            }
            // 接收到设备温湿度数据上报以后的处理
            else if(topic.contains("/user/tempHumUpload")) {
                JSONObject jsonObject = JSONObject.parseObject(content); // 字符串转json对象
                // 1. 组装温湿度数据，插入到device_prop_history_log表中
                DevicePropHistoryLog devicePropHistoryLog = new DevicePropHistoryLog();
                devicePropHistoryLog.setDeviceName(jsonObject.getString("deviceName"));//填入设备名称
                devicePropHistoryLog.setCurrentTemperature(jsonObject.getFloat("temperature"));// 填入当前温度
                devicePropHistoryLog.setCurrentHumidity(jsonObject.getFloat("humidity"));// 填入当前湿度
                devicePropHistoryLog.setGmtCreate(LocalDateTime.now());// 更新时间
                // save是mybatisplus提供的方法，可以直接调用，将数据存入到对应的表中
                devicePropHistoryLogService.save(devicePropHistoryLog);

                //2. 更新device_info表中的当前温湿度字段
                DeviceInfo deviceInfo = new DeviceInfo();
                deviceInfo.setDeviceName(jsonObject.getString("deviceName"));
                deviceInfo.setCurrentTemperature(jsonObject.getFloat("temperature"));
                deviceInfo.setCurrentHumidity(jsonObject.getFloat("humidity"));
                deviceInfo.setGmtUpdate(LocalDateTime.now());
                deviceInfoService.saveDeviceInfo(deviceInfo);
            }            // 接受到设备上报的报警数据
            else if (topic.contains("/user/tempAlarm")) {
                JSONObject jsonObject = JSONObject.parseObject(content);
                AlarmHistoryLog alarmHistoryLog = new AlarmHistoryLog();
                alarmHistoryLog.setDeviceName(jsonObject.getString("deviceName"));
                alarmHistoryLog.setAlarmTemperature(jsonObject.getFloat("alarm_temperature"));
                alarmHistoryLog.setTempThreshold(jsonObject.getFloat("temp_threshold"));
                alarmHistoryLog.setGmtCreate(LocalDateTime.now());   // 当前报警时间
                alarmHistoryLogService.save(alarmHistoryLog);

                DeviceInfo deviceInfo = new DeviceInfo();
                deviceInfo.setDeviceName(jsonObject.getString("deviceName"));
                deviceInfo.setAlarmStatus(true);
                deviceInfo.setLastAlarmDate(LocalDateTime.now());   // 当前报警时间
                deviceInfoService.saveDeviceInfo(deviceInfo);
            }



        } catch (Exception e) {
            System.out.println("processMessage occurs error "+ e);
        }
    }


    /**
     * 计算签名，password组装方法，请参见AMQP客户端接入说明文档。
     */
    private static String doSign(String toSignString, String secret, String signMethod) throws Exception {
        SecretKeySpec signingKey = new SecretKeySpec(secret.getBytes(), signMethod);
        Mac mac = Mac.getInstance(signMethod);
        mac.init(signingKey);
        byte[] rawHmac = mac.doFinal(toSignString.getBytes());
        return Base64.encodeBase64String(rawHmac);
    }

    private static JmsConnectionListener myJmsConnectionListener = new JmsConnectionListener() {
        /**
         * 连接成功建立。
         */
        @Override
        public void onConnectionEstablished(URI remoteURI) {
            System.out.println("onConnectionEstablished, remoteUri:{}"+ remoteURI);
        }

        /**
         * 尝试过最大重试次数之后，最终连接失败。
         */
        @Override
        public void onConnectionFailure(Throwable error) {
            System.out.println("onConnectionFailure, {}" + error.getMessage());
        }

        /**
         * 连接中断。
         */
        @Override
        public void onConnectionInterrupted(URI remoteURI) {
            System.out.println("onConnectionInterrupted, remoteUri:{}"+ remoteURI);
        }

        /**
         * 连接中断后又自动重连上。
         */
        @Override
        public void onConnectionRestored(URI remoteURI) {
            System.out.println("onConnectionRestored, remoteUri:{}"+  remoteURI);
        }

        @Override
        public void onInboundMessage(JmsInboundMessageDispatch envelope) {}

        @Override
        public void onSessionClosed(Session session, Throwable cause) {}

        @Override
        public void onConsumerClosed(MessageConsumer consumer, Throwable cause) {}

        @Override
        public void onProducerClosed(MessageProducer producer, Throwable cause) {}
    };
}
